package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.DnsCount;
import com.surfilter.massdata.spark.bean.SA7602;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;


public class DomainAnalyzTask_SA7602_dayNew implements ISparkTask{
	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7602_dayNew.class);

		@Override
		public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
			long start=System.currentTimeMillis();
			System.out.println("==============DomainAnalyzTask_SA7602 begin====================");
			List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
			try{
				DataFrame dataSource = dataFrames.get("dws_hour_sa7602");
				String sql="select domain,sum(traffic)as count  from dws_hour_sa7602_temp where  BUSS_TYPE='WEBSITE_RANK_SUM' and sta_range=1 GROUP BY domain";
				DataFrame df_sa7602 = dataSource.sqlContext().sql(sql);
				 JavaRDD<Map<String, String>> domainCounts = df_sa7602.toJavaRDD().map(new Function<Row,Map<String,String>>(){
					private static final long serialVersionUID = 1L;
						@Override
						public Map<String, String> call(Row row) throws Exception {
							Map<String, String> map=new HashMap<String,String>();
							String domain=row.getAs("domain").toString();
							String count=row.getAs("count").toString();
							map.put("domain", domain);
							map.put("count", count);
							return map;
						}
						
					});
	
		        JavaRDD<DnsCount> dnsRDD = domainCounts.map(new Function<Map<String,String>,DnsCount>() {
					private static final long serialVersionUID = 1L;
					@Override
					public DnsCount call(Map<String, String> map){
						DnsCount sa = new DnsCount();
						try{
							String top_domain = map.get("domain").toString();
							long count=Long.parseLong(map.get("count").toString());
							
							if(StringUtils.isNotBlank(top_domain)){
								 sa.setTop_domain(top_domain);
					    		 sa.setCount(count);
							}
						}catch(Exception e){
							if(log.isErrorEnabled()){
								log.error(e.getMessage(),e);
							}
						}
						return sa;
					}
				});
		    	
		    	DataFrame df_dc = dataSource.sqlContext().createDataFrame(dnsRDD, DnsCount.class);
		        df_dc.registerTempTable("df_dc");
				String joinsql = "select a.top_domain,a.count, "
						+ "(CASE WHEN ISNULL(b.license_key) = false THEN b.license_key ELSE 'N/A' END) AS license_key,"
						+ "(CASE WHEN ISNULL(b.site_name) = false THEN b.site_name ELSE 'N/A' END) AS site_name "
						+ "from df_dc a left join br2005_temp b on a.top_domain = b.top_domain order by a.count desc limit 100"; 

				DataFrame oper_tmp = dataSource.sqlContext().sql(joinsql);
				List<Row> top_domain_100 = oper_tmp.toJavaRDD().collect();
		    	List<SA7602> list_SA7602 = new ArrayList<SA7602>();	
		        String dayStr = commandMap.get("-d");
		        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
		        if(top_domain_100.size() > 0){
		        	int rank = 1;
		        	for(Row row_top_domain_100 : top_domain_100){
		        	    SA7602 sa = new SA7602();
		        		sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setBuss_type("WEBSITE_RANK_SUM");
		        		sa.setBuss_value("");
		        		sa.setRank(rank++);
		        		sa.setDomain(row_top_domain_100.getAs("top_domain").toString());
		        	    sa.setWebsite_name(row_top_domain_100.getAs("site_name").toString());
		        	    sa.setIcp_num(row_top_domain_100.getAs("license_key").toString());
		        	    sa.setTraffic(row_top_domain_100.getAs("count").toString());
		        		sa.setSta_range(1);
		        		list_SA7602.add(sa);
		        	}
		        }	
		        df_dc.sqlContext().dropTempTable("df_dc");
		    	JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
		    	JavaRDD<SA7602> rdd = ctx.parallelize(list_SA7602);
		    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA7602.class);
				outList.add(new OutQueueEntity("SA7602_TEMP", df));
				
				CommonUtils.deleteTaskTableData("SA7602", date, "WEBSITE_RANK_SUM", 1, "day");
			}catch(Exception e){
				if(log.isErrorEnabled()){
					log.error(e.getMessage(),e);
				}
			}
			long end=System.currentTimeMillis();
			double min=(end-start)*1.0/(1000*60);
			System.out.println("DomainAnalyzTask_SA7602_New:exectime: "+min+" min............");
			return outList;
		}
	}

